use std::{
    collections::{HashMap, HashSet},
    io,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
};

use rand::{thread_rng, Rng as _};
use tokio::sync::{mpsc, oneshot};

use crate::ws::resp::{ConnId, ConnUser, Msg, RoomId, SystemResp};

/// [`ChatServer`] 可接收的命令枚举
#[derive(Debug)]
enum Command {
    Connect {
        conn_tx: mpsc::UnboundedSender<Msg>,
        res_tx: oneshot::Sender<ConnId>,
    },

    Disconnect {
        conn: ConnId,
    },

    List {
        res_tx: oneshot::Sender<Vec<RoomId>>,
    },

    ListConnIds {
        res_tx: oneshot::Sender<Vec<ConnId>>,
    },

    ListUsers {
        res_tx: oneshot::Sender<Vec<ConnUser>>,
    },

    SetUser {
        res_tx: oneshot::Sender<()>,
        conn: ConnId,
        name: String,
    },

    Join {
        conn: ConnId,
        room: RoomId,
        res_tx: oneshot::Sender<()>,
    },

    Message {
        msg: Msg,
        conn: ConnId,
        res_tx: oneshot::Sender<()>,
    },

    Notice {
        msg: Msg,     // notice 广播内容
        room: RoomId, // notice 广播的房间
        res_tx: oneshot::Sender<()>,
    },
}

/// 链接管理和聊天房间逻辑
///
/// Call and spawn [`run`](Self::run) 以开始处理命令
#[derive(Debug)]
pub struct ChatServer {
    /// 用户连接ID与他们对应的消息接收器
    sessions: HashMap<ConnId, mpsc::UnboundedSender<Msg>>,

    /// 所有房间与房间内连接ID
    rooms: HashMap<RoomId, HashSet<ConnId>>,

    /// 所有用户ID与用户详细信息
    users: HashMap<ConnId, ConnUser>,

    /// 历史连接总数
    visitor_count: Arc<AtomicUsize>,

    /// 命令接收器
    cmd_rx: mpsc::UnboundedReceiver<Command>,
}

impl ChatServer {
    pub fn new() -> (Self, ChatServerHandle) {
        // 初始化房间，用户等集合
        let mut rooms = HashMap::with_capacity(4);

        // 新建默认房间 "main" 房间
        rooms.insert("main".to_owned(), HashSet::new());

        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();

        (
            Self {
                sessions: HashMap::new(),
                rooms,
                users: HashMap::new(),
                visitor_count: Arc::new(AtomicUsize::new(0)),
                cmd_rx,
            },
            ChatServerHandle { cmd_tx },
        )
    }

    /// 向一个房间内的用户发送消息
    pub async fn send_system_message(&self, room: &str, skip: ConnId, msg: impl Into<String>) {
        if let Some(sessions) = self.rooms.get(room) {
            let msg = msg.into();

            for conn_id in sessions {
                if *conn_id != skip {
                    if let Some(tx) = self.sessions.get(conn_id) {
                        // errors if client disconnected abruptly and hasn't been timed-out yet
                        let _ = tx.send(msg.clone());
                    }
                }
            }
        }
    }

    /// 向当前房间内所有其他用户发送消息
    ///
    /// `conn`用于查找当前房间，并防止连接发送的消息也被自己接收
    async fn send_message(&self, conn: ConnId, msg: impl Into<String>) {
        if let Some(room) = self
            .rooms
            .iter()
            .find_map(|(room, participants)| participants.contains(&conn).then_some(room))
        {
            self.send_system_message(room, conn, msg).await;
        };
    }

    /// 注册新的 session
    async fn connect(&mut self, tx: mpsc::UnboundedSender<Msg>) -> ConnId {
        // log::info!("Someone joined");

        // 通知主房间内全部用户
        self.send_system_message("main", 0, "someone joined").await;

        // 使用随机 connection ID 注册 session，例如: 7212139879744255790
        let id = thread_rng().gen::<usize>();
        self.sessions.insert(id, tx);
        self.users.insert(id, ConnUser::new(id));

        // 自动将 session 加入至房间
        self.rooms
            .entry("main".to_owned())
            .or_insert_with(HashSet::new)
            .insert(id);

        self.online_info().await;

        id
    }

    async fn online_info(&mut self) {
        let count = self.visitor_count.fetch_add(1, Ordering::SeqCst);
        let online = self.sessions.len();

        // 访客在线数
        let online_count_resp: SystemResp<usize> = SystemResp {
            event: "system".to_owned(),
            kind: "online".to_owned(),
            data: vec![],
            number: online,
        };
        self.send_system_message(
            "main",
            0,
            serde_json::to_string(&online_count_resp).unwrap(),
        )
        .await;

        // 访客总数
        let total_count_resp: SystemResp<usize> = SystemResp {
            event: "system".to_owned(),
            kind: "total".to_owned(),
            data: vec![],
            number: count,
        };
        self.send_system_message("main", 0, serde_json::to_string(&total_count_resp).unwrap())
            .await;
    }

    /// 从房间中注销 session 并断开广播链接
    async fn disconnect(&mut self, conn_id: ConnId) {
        let mut rooms: Vec<String> = Vec::new();

        // remove sender
        if self.sessions.remove(&conn_id).is_some() {
            // remove user data from users
            self.users.remove(&conn_id);
            // remove session from all rooms
            for (name, sessions) in &mut self.rooms {
                if sessions.remove(&conn_id) {
                    rooms.push(name.to_owned());
                }
            }
        }

        // send message to other users
        for room in rooms {
            self.send_system_message(&room, 0, "someone disconnected")
                .await;
        }

        self.online_info().await;
    }

    /// 返回房间名称列表
    fn list_rooms(&mut self) -> Vec<String> {
        // self.rooms.keys().cloned().collect()
        let mut res: Vec<String> = vec![];
        for (key, value) in self.rooms.iter() {
            let vs: Vec<String> = value.iter().map(|v| format!("{}", v)).collect();
            let vstr = vs.join(",");
            res.push(format!("{}:{}", key, vstr));
        }
        res
    }

    /// 返回用户ID列表
    fn list_connids(&mut self) -> Vec<usize> {
        self.sessions.keys().cloned().collect()
    }

    /// 返回用户信息列表
    fn list_users(&mut self) -> Vec<ConnUser> {
        self.users.values().cloned().collect()
    }

    /// 设置用户名至 server users 中，共享
    fn set_user(&mut self, conn_id: ConnId, name: String) {
        let user = self.users.get_mut(&conn_id);
        match user {
            Some(u) => u.name = name,
            None => {}
        }
    }

    /// 加入新房间，同时发送断开消息给旧房间
    async fn join_room(&mut self, conn_id: ConnId, room: String) {
        let mut rooms = Vec::new();

        for (n, sessions) in &mut self.rooms {
            if sessions.remove(&conn_id) {
                rooms.push(n.to_owned());
            }
        }

        for room in rooms {
            self.send_system_message(&room, 0, "someone disconnected room")
                .await;
        }

        self.rooms
            .entry(room.clone())
            .or_insert_with(HashSet::new)
            .insert(conn_id);

        self.send_system_message(&room, conn_id, "someone connected room")
            .await;
    }

    pub async fn run(mut self) -> io::Result<()> {
        while let Some(cmd) = self.cmd_rx.recv().await {
            match cmd {
                Command::Connect { conn_tx, res_tx } => {
                    let conn_id = self.connect(conn_tx).await;
                    let _ = res_tx.send(conn_id);
                }

                Command::Disconnect { conn } => {
                    self.disconnect(conn).await;
                }

                Command::List { res_tx } => {
                    let _ = res_tx.send(self.list_rooms());
                }

                Command::ListConnIds { res_tx } => {
                    let _ = res_tx.send(self.list_connids());
                }

                Command::ListUsers { res_tx } => {
                    let _ = res_tx.send(self.list_users());
                }

                Command::SetUser { res_tx, conn, name } => {
                    self.set_user(conn, name);
                    let _ = res_tx.send(());
                }

                Command::Join { conn, room, res_tx } => {
                    self.join_room(conn, room).await;
                    let _ = res_tx.send(());
                }

                Command::Message { conn, msg, res_tx } => {
                    self.send_message(conn, msg).await;
                    let _ = res_tx.send(());
                }

                // 发送 notice 至 main（主房间）内所有客户端
                Command::Notice { msg, room, res_tx } => {
                    self.send_system_message(&room, 0, msg).await;
                    let _ = res_tx.send(());
                }
            }
        }

        Ok(())
    }
}

/// 命令处理器
#[derive(Debug, Clone)]
pub struct ChatServerHandle {
    cmd_tx: mpsc::UnboundedSender<Command>,
}

impl ChatServerHandle {
    pub async fn connect(&self, conn_tx: mpsc::UnboundedSender<String>) -> ConnId {
        let (res_tx, res_rx) = oneshot::channel();

        // unwrap: chat server should not have been dropped
        self.cmd_tx
            .send(Command::Connect { conn_tx, res_tx })
            .unwrap();

        // unwrap: chat server does not drop out response channel
        res_rx.await.unwrap()
    }

    /// 获取所有已创建的房间
    pub async fn list_rooms(&self) -> Vec<String> {
        let (res_tx, res_rx) = oneshot::channel();
        self.cmd_tx.send(Command::List { res_tx }).unwrap();
        res_rx.await.unwrap()
    }

    pub async fn list_connids(&self) -> Vec<usize> {
        let (res_tx, res_rx) = oneshot::channel();
        self.cmd_tx.send(Command::ListConnIds { res_tx }).unwrap();
        res_rx.await.unwrap()
    }

    pub async fn list_users(&self) -> Vec<ConnUser> {
        let (res_tx, res_rx) = oneshot::channel();
        self.cmd_tx.send(Command::ListUsers { res_tx }).unwrap();
        res_rx.await.unwrap()
    }

    pub async fn set_user(&self, conn: ConnId, name: String) {
        let (res_tx, res_rx) = oneshot::channel();
        self.cmd_tx
            .send(Command::SetUser { res_tx, conn, name })
            .unwrap();
        res_rx.await.unwrap()
    }

    /// 加入房间，不存在则新建房间
    pub async fn join_room(&self, conn: ConnId, room: impl Into<String>) {
        let (res_tx, res_rx) = oneshot::channel();
        let room_id = room.into();
        log::info!("connId: {conn} joined room: {}", room_id);
        self.cmd_tx
            .send(Command::Join {
                conn,
                room: room_id,
                res_tx,
            })
            .unwrap();
        res_rx.await.unwrap();
    }

    // 向所有客户端发送 Notice 通知
    pub fn send_notice(&self, room: &str, msg: impl Into<String>) {
        let (res_tx, res_rx) = oneshot::channel();
        // unwrap: chat server 不应被删除
        self.cmd_tx
            .send(Command::Notice {
                msg: msg.into(),
                room: room.into(),
                res_tx,
            })
            .unwrap();
        // chat server 服务器不会丢弃我们的响应通道
        drop(res_rx);
    }

    /// 广播消息到当前房间
    pub async fn send_message(&self, conn: ConnId, msg: impl Into<String>) {
        let (res_tx, res_rx) = oneshot::channel();
        self.cmd_tx
            .send(Command::Message {
                msg: msg.into(),
                conn,
                res_tx,
            })
            .unwrap();
        res_rx.await.unwrap();
    }

    pub fn disconnect(&self, conn: ConnId) {
        self.cmd_tx.send(Command::Disconnect { conn }).unwrap();
    }
}
